package ApplicationStart.SparkService;


import ApplicationStart.Configuration.SparkDefaultConfig;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.SparkContext$;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.Milliseconds;
import org.apache.spark.streaming.StreamingContext;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;

@Service  //可以将bean 注入到其中
public class SparkService {


    private SparkConf sparkConf;

    private SparkContext sc;

    @Autowired
    public SparkService(SparkDefaultConfig sparkDefaultConfig) {
        this.sparkConf = new SparkConf().setAppName(sparkDefaultConfig.getAppname())
                .setMaster(sparkDefaultConfig.getMaster())
                .setJars(sparkDefaultConfig.getJarpath()); //提交任务到远程
        initSC();
    }

    //初始化
    private void initSC(){
        this.sc = SparkContext$.MODULE$.getOrCreate(this.sparkConf);
    }


    //获取scala SparkContext
    @Bean("sparkContextFromSparkService")
    public SparkContext getSC() {
        return this.sc;
    }

    //获取SparkSession
    @Bean("sparkSessionFromSparkService")
    public SparkSession getSparkSQL() {
        return SparkSession.builder().sparkContext(sc).getOrCreate();
    }

    //获取 Java Spark Context
    @Bean("JavaSparkContextFromSparkService")
    public JavaSparkContext getJavaSC(){
        return JavaSparkContext.fromSparkContext(sc);
    }

    //获取java stream
    @Bean("JavaStreamingContextFromSparkService")
    public JavaStreamingContext getJavaStream(){
       return new JavaStreamingContext(getJavaSC(), Milliseconds.apply(5000));
    }

    //获取scala stream
    @Bean("StreamingContextFromSparkService")
    public StreamingContext getStream(){
        return getJavaStream().ssc();
    }

    public SparkConf getSparkConf(){
        return sparkConf;
    }
}
